package com.codejiwei.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/5/25 14:17
 */
public class Flink_SQL_Kafka2Kafka {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(10);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 1. 注册SourceTable: source_sensor
//        tableEnv.executeSql("create table source_sensor (id string, ts bigint, vc int) with("
//                + "'connector' = 'kafka',"
//                + "'topic' = 'topic_source_sensor',"
//                + "'properties.bootstrap.servers' = 'localhost:9092',"
//                + "'properties.group.id' = 'jiwei',"
//                + "'scan.startup.mode' = 'latest-offset',"
//                + "'format' = 'json'"
//                + ")");

        tableEnv.executeSql("CREATE TABLE KafkaTable (\n" +
                "  `user_id` BIGINT,\n" +
                "  `item_id` BIGINT,\n" +
                "  `behavior` STRING,\n" +
                "  `ts` TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'user_behavior',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'csv'\n" +
                ")");


//        // 2. 注册SinkTable: sink_sensor
//        tableEnv.executeSql("create table sink_sensor(id string, ts bigint, vc int) with("
//                + "'connector' = 'kafka',"
//                + "'topic' = 'topic_sink_sensor',"
//                + "'properties.bootstrap.servers' = 'localhost:9092',"
//                + "'format' = 'json'"
//                + ")");

        tableEnv.executeSql("create table sink_sensor with ('connector' = 'print') like KafkaTable (excluding all)");


        // 3. 从SourceTable 查询数据, 并写入到 SinkTable
        tableEnv.executeSql("insert into sink_sensor select * from KafkaTable");

    }
}
